package com.zhengb.flink.study.sql.streamsql;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.StreamTableEnvironment;

import java.util.Arrays;

/**
 * dataStream 简单sql
 *
 * @author zhengbo
 * @date 2019/12/31
 */
@Slf4j
public class StreamSqlExample {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        DataStreamSource<Order> orderA = env.fromCollection(Arrays.asList(
                new Order(1L, "beer", 3),
                new Order(2L, "cococola", 4),
                new Order(3L, "rubber", 2)));

        DataStreamSource<Order> orderB = env.fromCollection(Arrays.asList(
                new Order(1L, "pen", 3),
                new Order(2L, "pencil", 3),
                new Order(3L, "beer", 1)));

        //Table tableA = tableEnvironment.fromDataStream(orderA, "user,product,amount");

        tableEnvironment.registerDataStream("OrderA", orderA, "user,product,amount");

        tableEnvironment.registerDataStream("OrderB", orderB, "user,product,amount");

        Table result = tableEnvironment.sqlQuery("select * from " + " OrderA " + " where amount>2 " +
                " UNION all select * from OrderB where amount <2");


        tableEnvironment.toAppendStream(result, Order.class).addSink(new SinkFunction<Order>() {

            @Override
            public void invoke(Order value, Context context) throws Exception {
                log.info("sink data:{}", JSONObject.toJSONString(value));
            }
        });

        env.execute("StreamSqlExample");
    }
}
